<?php
/**
 * Created by PhpStorm.
 * @description:
 * @time: 2018年5月24日21:25:22
 * @Author: yfl
 * @QQ 554665488
 * Date: 2018-5-24
 * Time: 21:03
 */

/**
 * 2.进程模块简单封装
 * 3.进程模块异常处理任务重新投递
 * 4.主进程终止子进程处理方式
 */
class Pool
{
    protected $table;
    protected $mpid;//进程id

    public function __construct()
    {
        $this->mpid = getmypid();
        $this->table = $this->create_table();  //将进程的pid和任务的内容进行绑定
//        var_export($this->table);
//        var_dump($this->mpid);
//        var_dump(posix_getpid());
        $this->run();
        $this->signal();//信号监听
    }

    //捕获子进程结束的时候的信号，回收子进程
    public function signal()
    {
        swoole_process::signal(SIGCHLD, function ($sig) {
            while ($res = swoole_process::wait(false)) {
                if ($res['code'] > 0) { //出现非0状态
                    $pid = $res['pid']; //子进程结束的pid
                    $data = $this->table->get($pid);
                    $this->table->del($pid);
                    if ($data['attempts'] == 0) {
                        $process=$this->create_process();//重新拉起进程产生新的pid
                        $this->table->set($process->pid,['data'=>$data,'attempts'=>$data['attempts']+1]);//新创建的进程
                        $process->write($data['data']);
                        return;
                    }
                    //file_put_contents(); 记录日志|发送邮件|异常队列当中
                }
            }
        });
    }

    //获取任务 创建进程
    protected function run()
    {
        $data = [
            '990979940@qq.com',
            '213123@qq.com',
            '457567657@qq.com'
        ];
        foreach ($data as $v) {
            $process = $this->create_process();
//            var_dump($process);
            $this->table->set($process->pid, ['data' => $v]);
            $process->write($v);//向子进程投递数据任务
//            var_dump($process->read()); //读取子进程向主进程投递的数据 使用异步读取数据的时候 这里获取不到数据
        }

    }

    //创建一个内存表
    protected function create_table()
    {
        $table = new swoole\table(1024);
        $table->column('attempts', swoole\table::TYPE_INT, 2);
        $table->column('data', swoole\table::TYPE_STRING, 64);
        $table->create();
        return $table;
    }


    //创建子进程
    protected function create_process()
    {
        $process = new swoole\Process([$this, 'callback_function']);
        $process->start();//启动子进程

        //echo $process->read().PHP_EOL; //同步阻塞

        //读事件监听,读取内容变成异步读取,在事件触发时才会调用
        //异步读取子进程向写入的管道数据
        swoole_event::add($process->pipe, function () use ($process) {
            $data = json_decode($process->read(), true);
//            var_dump($data);
            if ($data['status'] === false) {
                //执行任务失败动作处理 2018年5月25日00:23:39
            } else {
                swoole_event::del($process->pipe);
                echo '子进程发送的数据' . PHP_EOL;
                var_dump($data);
            }
        });

        return $process;
    }

    //子进程处理逻辑
    public function callback_function($worker)
    {
        $read_res = $worker->read() . '====';//这里开始子进程干活的动作
//        var_dump($read_res);
        //模拟子进程执行任务失败
        $rand = mt_rand(1, 200);
        for ($i = 0; $i < 20; $i++) {
//            var_dump($worker);
            echo '子进程在执行任务' . $read_res . PHP_EOL;
            $this->checkPid($worker, $this->mpid);//检测主进程是否异常终止避免子进程成为僵尸进程
            sleep(2);
        }
        if ($rand == 1) {
            throw new Exception('出现了异常');
            //执行发短信的逻辑
        }

        $data = [
            'status' => true, 'read_res' => $read_res, 'msg' => '任务执行成功'
        ];
        $worker->write(json_encode($data));//写入到主进程
//        var_dump($worker->write(json_encode($data)));
    }

    public function checkPid($worker, $mpid)
    {
        //等待子进程任务处理完毕之后再去关闭，根据业务场景 如果需要跟主进程交互的 主进程关闭了
        //子进程需要关闭。如果不需要的 等待子进程干完活后再去关闭
        if (!swoole\Process::kill($mpid, 0)) {
            $msg = "主进程已经退出，工作进程{$worker->pid}退出";
            echo $msg . PHP_EOL;
            $worker->exit();

        }
    }
}

new pool();